package com.larry.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession, TypedColumn, expressions, functions}

object Sql_Oper_UDF_3 {

  def main(args: Array[String]): Unit = {
    //TODO  使用spark coalesce  缩减分区
    //TODO  默认情况下 缩减分区不会shuffle

    val conf = new SparkConf().setMaster("local[*]").setAppName("sql")

    //创建session对象
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    import spark.implicits._

    //读取json
    val df = spark.read.json("input/user.json")

    df.createOrReplaceTempView("user")
    val myAvgUDAF = new MyAvgUDAF2

    spark.udf.register("myavg",functions.udaf(myAvgUDAF))

    spark.sql("select myavg(age) from user").show()
    //关闭资源
    spark.stop()

  }
}
class MyAvgUDAF2 extends Aggregator[Long,AgeBuffer,Double]{
  override def zero: AgeBuffer = AgeBuffer(0L,0L)

  override def reduce(b: AgeBuffer, a: Long): AgeBuffer = {
    b.sum += a
    b.count += 1
    b
  }

  override def merge(b1: AgeBuffer, b2: AgeBuffer): AgeBuffer = {
    b1.sum = b1.sum + b2.sum
    b1.count = b1.count + b2.count
    b1
  }

  override def finish(reduction: AgeBuffer): Double = {
    reduction.sum.toDouble / reduction.count
  }

  override def bufferEncoder: Encoder[AgeBuffer] = Encoders.product

  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

//缓存类型
//case class AgeBuffer(var sum:Long,var count:Long)
